
/*
* 数据统计类:
* 足球:
*   赛季-竞彩玩法赛果统计
* */

//
//cp FirstSparkApp/out/artifacts/FirstSparkAppJar/firstsparkapp.jar /data/caiqiu/prediction/jar/fb_FBST2017003.jar
//
//scp FirstSparkApp/out/artifacts/FirstSparkAppJar/firstsparkapp.jar root@172.16.0.71:/root/fb_FBST2017003.jar
//
//nohup ./bin/spark-submit --master spark://skn-pmukvrk0-spark-master:7077 --class caiqr.model.statistics.fb.FBST2017003 --jars /usr/local/spark/lib/mysql-connector-java-5.1.35.jar --executor-memory 4G /root/fb_FBST2017003.jar save_db_info=172.16.4.17-prediction-caiqiu-Caiqiu502 maxResultSize=4g job_id=8439 spark_id=11 big_file=hdfs://skn-qcqegnt5-hadoop-master:9000/data/caiqiu/csv/fb_match_500w_rq_1.csv  output_file=hdfs://skn-qcqegnt5-hadoop-master:9000/data/caiqiu/result/FBST2017003.csv  > /root/a.log < /dev/null 2>&1 &


//
//sqoop export  --connect jdbc:mysql://172.16.4.17/prediction --username root --password Caiqiu502 --table FBST2017003 --update-mode allowinsert --update-key "season_id" --fields-terminated-by ','  -export-dir hdfs://skn-qcqegnt5-hadoop-master:9000/data/caiqiu/result/FBST2017003.csv
//


//caiqr.model.statistics.fb.FBST2017003

package caiqr.model.statistics.fb

import com.redislabs.provider.redis._
import caiqr.utils.{AllAsiaInputFile, PredictionUtils, PredictionDBUtils, AllFBMatchInputFile}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, DataFrame}
import java.text.SimpleDateFormat
import java.sql.DriverManager
import java.lang.IllegalArgumentException

object FBST2017003 {

  def main(args: Array[String]){

    //////////////////////////////// 接收参数 ////////////////////////////////
    // 将参数转换为 Map
    val cmd_list = args.toList
    val cmd = cmd_list.map{p =>
      val items = p.split("=")
      (items(0),items(1))
    }
    val cmd_map = cmd.toMap


    val big_file_res1 = cmd_map.get("big_file") //赛事文件
    val big_file = big_file_res1.getOrElse("")

    val save_db_info_res1 = cmd_map.get("save_db_info")
    val save_db_info = save_db_info_res1.getOrElse("")

    val maxResultSize_res1 = cmd_map.get("maxResultSize")
    val maxResultSize = maxResultSize_res1.getOrElse("4g")

    val job_id_res1 = cmd_map.get("job_id")
    val job_id = job_id_res1.getOrElse("")

    val model_id_res1 = cmd_map.get("spark_id")
    val model_id = model_id_res1.getOrElse("")

    val output_file = cmd_map.get("output_file").getOrElse("") //输出文件


    // 1. 环境变量
    val sc = new SparkContext(new SparkConf()
      .setAppName("FBST2017003")
      .set("spark.driver.maxResultSize", maxResultSize)
    )
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)


    val match_df = AllFBMatchInputFile.load_rq_match_file(sc, sqlContext, big_file)
    //match_df.collect().take(100).foreach(println)
    val src_match_df = match_df.orderBy("season_id").select("season_id","result","rq_result","goal","half_all_result","new_result")


    val match_rdd = src_match_df.rdd
      .map( p => ((p.getString(0)),(p.getString(1),p.getString(2),p.getString(3),p.getString(4),p.getString(5))) )
      .groupByKey().map( p => (p._1, p._2.toArray))


    val total_match_rdd = match_rdd.map( p =>
      (p._1, (p._2.map(p => p._1).reduce(_+"_"+_),
              p._2.map(p => p._2).reduce(_+"_"+_),
              p._2.map(p => p._3).reduce(_+"_"+_),
              p._2.map(p => p._4).reduce(_+"_"+_),
              p._2.map(p => p._5).reduce(_+"_"+_))
        )
    ).map { p =>
      val result = p._2._1.toString.split("_")
      val rq = p._2._2.split("_")
      val goal = p._2._3.split("_")
      val half_all = p._2._4.split("_")
      val score = p._2._5.split("_")
      (p._1, result, rq, goal, half_all, score)
    }
//    total_match_rdd.collect().take(10).foreach{p =>
//      println(p._1)
//      println(p._2.mkString(","))
//      println(p._3.mkString(","))
//      println(p._4.mkString(","))
//      println(p._5.mkString(","))
//      println(p._6.mkString(","))
//    }

    // 分别计算各个赛果比例
    //OUT:
    //(3,0.5)(1,0.25)(0,0.25)
    //(3,0.47)(1,0.18)(0,0.35)
    //(0,0.09)(1,0.2)(2,0.27)(3,0.14)(4,0.16)(5,0.07)(6,0.04)(7,0.03)
    //(00,0.16)(01,0.06)(03,0.02)(10,0.1)(11,0.16)(13,0.18)(30,0.0)(31,0.03)(33,0.3)
    //(00,0.09)(11,0.12)(22,0.04)(33,0.01)(99,0.0)(10,0.12)(20,0.09)(21,0.05)(30,0.04)(31,0.05)(32,0.03)(40,0.04)(41,0.02)(42,0.02)(50,0.01)(51,0.01)(52,0.0)(90,0.03)(01,0.07)(02,0.06)(12,0.03)(03,0.02)(13,0.02)(23,0.01)(04,0.01)(14,0.01)(24,0.0)(05,0.01)(15,0.0)(25,0.0)(09,0.02)
    val ret_match_rdd = total_match_rdd.map{ p =>

      // 1.胜平负
      val results = p._2
      val result_cnt = results.length.toInt //让球胜平负赛果
      val win = results.count(_ == "3")
      val draw = results.count(_ == "1")
      val loss = results.count(_ == "0")
      val spf_3 = f"${win.toDouble/result_cnt}%1.2f".toDouble
      val spf_1 = f"${draw.toDouble/result_cnt}%1.2f".toDouble
      val spf_0 = f"${loss.toDouble/result_cnt}%1.2f".toDouble
      val spf = Array(("3",spf_3),("1",spf_1),("0",spf_0))


      // 2.让球胜平负
      val rqresults = p._3
      val rqresult_cnt = rqresults.length.toInt //让球胜平负赛果
      val rqwin = rqresults.count(_ == "3")
      val rqdraw = rqresults.count(_ == "1")
      val rqloss = rqresults.count(_ == "0")
      val rqspf_3 = f"${rqwin.toDouble/rqresult_cnt}%1.2f".toDouble
      val rqspf_1 = f"${rqdraw.toDouble/rqresult_cnt}%1.2f".toDouble
      val rqspf_0 = f"${rqloss.toDouble/rqresult_cnt}%1.2f".toDouble
      val rqspf = Array(("3",rqspf_3),("1",rqspf_1),("0",rqspf_0))


      // 3.总进球
      val gresults = p._4
      val gresult_cnt = gresults.length.toInt //总进球
      val g0 = gresults.count(_ == "0")
      val g1 = gresults.count(_ == "1")
      val g2 = gresults.count(_ == "2")
      val g3 = gresults.count(_ == "3")
      val g4 = gresults.count(_ == "4")
      val g5 = gresults.count(_ == "5")
      val g6 = gresults.count(_ == "6")
      val g7 = gresults.count(_ == "7")
      val rg0 = f"${g0.toDouble/gresult_cnt}%1.2f".toDouble
      val rg1 = f"${g1.toDouble/gresult_cnt}%1.2f".toDouble
      val rg2 = f"${g2.toDouble/gresult_cnt}%1.2f".toDouble
      val rg3 = f"${g3.toDouble/gresult_cnt}%1.2f".toDouble
      val rg4 = f"${g4.toDouble/gresult_cnt}%1.2f".toDouble
      val rg5 = f"${g5.toDouble/gresult_cnt}%1.2f".toDouble
      val rg6 = f"${g6.toDouble/gresult_cnt}%1.2f".toDouble
      val rg7 = f"${g7.toDouble/gresult_cnt}%1.2f".toDouble
      val goal = Array(("0",rg0),("1",rg1),("2",rg2),("3",rg3),("4",rg4),("5",rg5),("6",rg6),("7",rg7))


      // 4.半全场
      val hresults = p._5
      val hresult_cnt = hresults.length.toInt
      val h00 = hresults.count(_ == "00")
      val h01 = hresults.count(_ == "01")
      val h03 = hresults.count(_ == "03")
      val h10 = hresults.count(_ == "10")
      val h11 = hresults.count(_ == "11")
      val h13 = hresults.count(_ == "13")
      val h30 = hresults.count(_ == "30")
      val h31 = hresults.count(_ == "31")
      val h33 = hresults.count(_ == "33")
      val rh00 = f"${h00.toDouble/hresult_cnt}%1.2f".toDouble
      val rh01 = f"${h01.toDouble/hresult_cnt}%1.2f".toDouble
      val rh03 = f"${h03.toDouble/hresult_cnt}%1.2f".toDouble
      val rh10 = f"${h10.toDouble/hresult_cnt}%1.2f".toDouble
      val rh11 = f"${h11.toDouble/hresult_cnt}%1.2f".toDouble
      val rh13 = f"${h13.toDouble/hresult_cnt}%1.2f".toDouble
      val rh30 = f"${h30.toDouble/hresult_cnt}%1.2f".toDouble
      val rh31 = f"${h31.toDouble/hresult_cnt}%1.2f".toDouble
      val rh33 = f"${h33.toDouble/hresult_cnt}%1.2f".toDouble

      val half = Array(
        ("00",rh00),("01",rh01),("03",rh03),
        ("10",rh10),("11",rh11),("13",rh13),
        ("30",rh30),("31",rh31),("33",rh33)
      )



      //5. 比分
      val sresult = p._6
      val sresult_cnt = sresult.length.toInt
      val s00 = sresult.count(_ == "00")
      val s11 = sresult.count(_ == "11")
      val s22 = sresult.count(_ == "22")
      val s33 = sresult.count(_ == "33")
      val s99 = sresult.count(_ == "99")
      val rs00 = f"${s00.toDouble/sresult_cnt}%1.2f".toDouble
      val rs11 = f"${s11.toDouble/sresult_cnt}%1.2f".toDouble
      val rs22 = f"${s22.toDouble/sresult_cnt}%1.2f".toDouble
      val rs33 = f"${s33.toDouble/sresult_cnt}%1.2f".toDouble
      val rs99 = f"${s99.toDouble/sresult_cnt}%1.2f".toDouble

      val s10 = sresult.count(_ == "10")
      val s20 = sresult.count(_ == "20")
      val s21 = sresult.count(_ == "21")
      val s30 = sresult.count(_ == "30")
      val s31 = sresult.count(_ == "31")
      val s32 = sresult.count(_ == "32")
      val s40 = sresult.count(_ == "40")
      val s41 = sresult.count(_ == "41")
      val s42 = sresult.count(_ == "42")
      val s50 = sresult.count(_ == "50")
      val s51 = sresult.count(_ == "51")
      val s52 = sresult.count(_ == "52")
      val s90 = sresult.count(_ == "90")
      val rs10 = f"${s10.toDouble/sresult_cnt}%1.2f".toDouble
      val rs20 = f"${s20.toDouble/sresult_cnt}%1.2f".toDouble
      val rs21 = f"${s21.toDouble/sresult_cnt}%1.2f".toDouble
      val rs30 = f"${s30.toDouble/sresult_cnt}%1.2f".toDouble
      val rs31 = f"${s31.toDouble/sresult_cnt}%1.2f".toDouble
      val rs32 = f"${s32.toDouble/sresult_cnt}%1.2f".toDouble
      val rs40 = f"${s40.toDouble/sresult_cnt}%1.2f".toDouble
      val rs41 = f"${s41.toDouble/sresult_cnt}%1.2f".toDouble
      val rs42 = f"${s42.toDouble/sresult_cnt}%1.2f".toDouble
      val rs50 = f"${s50.toDouble/sresult_cnt}%1.2f".toDouble
      val rs51 = f"${s51.toDouble/sresult_cnt}%1.2f".toDouble
      val rs52 = f"${s52.toDouble/sresult_cnt}%1.2f".toDouble
      val rs90 = f"${s90.toDouble/sresult_cnt}%1.2f".toDouble


      val s01 = sresult.count(_ == "01")
      val s02 = sresult.count(_ == "02")
      val s12 = sresult.count(_ == "12")
      val s03 = sresult.count(_ == "03")
      val s13 = sresult.count(_ == "13")
      val s23 = sresult.count(_ == "23")
      val s04 = sresult.count(_ == "04")
      val s14 = sresult.count(_ == "14")
      val s24 = sresult.count(_ == "24")
      val s05 = sresult.count(_ == "05")
      val s15 = sresult.count(_ == "15")
      val s25 = sresult.count(_ == "25")
      val s09 = sresult.count(_ == "09")
      val rs01 = f"${s01.toDouble/sresult_cnt}%1.2f".toDouble
      val rs02 = f"${s02.toDouble/sresult_cnt}%1.2f".toDouble
      val rs12 = f"${s12.toDouble/sresult_cnt}%1.2f".toDouble
      val rs03 = f"${s03.toDouble/sresult_cnt}%1.2f".toDouble
      val rs13 = f"${s13.toDouble/sresult_cnt}%1.2f".toDouble
      val rs23 = f"${s23.toDouble/sresult_cnt}%1.2f".toDouble
      val rs04 = f"${s04.toDouble/sresult_cnt}%1.2f".toDouble
      val rs14 = f"${s14.toDouble/sresult_cnt}%1.2f".toDouble
      val rs24 = f"${s24.toDouble/sresult_cnt}%1.2f".toDouble
      val rs05 = f"${s05.toDouble/sresult_cnt}%1.2f".toDouble
      val rs15 = f"${s15.toDouble/sresult_cnt}%1.2f".toDouble
      val rs25 = f"${s25.toDouble/sresult_cnt}%1.2f".toDouble
      val rs09 = f"${s09.toDouble/sresult_cnt}%1.2f".toDouble

      val score = Array(
        ("00",rs00),("11",rs11),("22",rs22),("33",rs33),("99",rs99),
        ("10",rs10),("20",rs20),("21",rs21),
        ("30",rs30),("31",rs31),("32",rs32),
        ("40",rs40),("41",rs41),("42",rs42),
        ("50",rs50),("51",rs51),("52",rs52),("90",rs90),
        ("01",rs01),("02",rs02),("12",rs12),
        ("03",rs03),("13",rs13),("23",rs23),
        ("04",rs04),("14",rs14),("24",rs24),
        ("05",rs05),("15",rs15),("25",rs25),("09",rs09)
      )

      (p._1, spf, rqspf, goal, half, score)
    }
//    ret_match_rdd.collect().take(10).foreach { p =>
//      println(p._1)
//      p._2.foreach(print)
//      println("")
//      p._3.foreach(print)
//      println("")
//      p._4.foreach(print)
//      println("")
//      p._5.foreach(print)
//      println("")
//      p._6.foreach(print)
//      println("")
//    }


    // 选项概率排序(Desc),比分获取前10个.
    val final_ret_match_rdd = ret_match_rdd.map{ p =>
      val spf = p._2.sortWith{case(res1,res2) => res1._2 > res2._2}.map(p => s"${p._1}:${p._2}").mkString(";")
      val rqspf = p._3.sortWith{case(res1,res2) => res1._2 > res2._2}.map(p => s"${p._1}:${p._2}").mkString(";")
      val goal = p._4.sortWith{case(res1,res2) => res1._2 > res2._2}.map(p => s"${p._1}:${p._2}").mkString(";")
      val half = p._5.sortWith{case(res1,res2) => res1._2 > res2._2}.map(p => s"${p._1}:${p._2}").mkString(";")
      val score = p._6.sortWith{case(res1,res2) => res1._2 > res2._2}.take(10).map(p => s"${p._1}:${p._2}").mkString(";")
      Array(p._1, spf, rqspf, goal, half, score).mkString(",")
    }
    //final_ret_match_rdd.collect().take(10).foreach(println)


    //保存结果到 HDFS
    PredictionUtils.save_result_to_hdfs(final_ret_match_rdd, output_file)



    // 更新job 和spark状态
    // spark运行完成,待 sqoop导入DB
    PredictionDBUtils.update_job_spark_status(save_db_info, job_id, model_id)


    sc.stop()
  }


}





